/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 */
/* -------------------------------------------------------------------------
 *
 * imcs_ctlg.cpp
 * 	  openGauss define and remove imcs code.
 *
 * Portions Copyright (c) 2020 Huawei Technologies Co.,Ltd.
 * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 * Portions Copyright (c) 2010-2012 Postgres-XC Development Group
 * Portions Copyright (c) 2021, openGauss Contributors
 *
 *
 * IDENTIFICATION
 * 	  src/gausskernel/optimizer/commands/imcs_ctlg.cpp
 *
 * -------------------------------------------------------------------------
 */
#include "c.h"
#include "postgres.h"
#include "knl/knl_variable.h"

#include <cstddef>
#include "access/cstore_delta.h"
#include "access/reloptions.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/index.h"
#include "catalog/indexing.h"
#include "catalog/pg_constraint.h"
#include "catalog/pg_opclass.h"
#include "catalog/pg_opfamily.h"
#include "catalog/pg_partition.h"
#include "catalog/pg_partition_fn.h"
#include "catalog/pg_tablespace.h"
#include "catalog/pg_type.h"
#include "catalog/gs_db_privilege.h"
#include "catalog/namespace.h"
#include "catalog/pg_namespace.h"
#include "commands/comment.h"
#include "commands/dbcommands.h"
#include "commands/defrem.h"
#include "catalog/heap.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "foreign/foreign.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
#include "optimizer/clauses.h"
#include "optimizer/planner.h"
#include "parser/parse_coerce.h"
#include "parser/parse_func.h"
#include "parser/parse_oper.h"
#ifdef PGXC
#include "optimizer/pgxcship.h"
#include "parser/parse_utilcmd.h"
#include "pgxc/pgxc.h"
#endif
#include <vector>
#include <algorithm>
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/tcap.h"
#include "tcop/utility.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/palloc.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "access/heapam.h"
#include "access/ustore/knl_whitebox_test.h"
#include "securec.h"
#include "catalog/gs_imcs.h"
#include "access/imcs/imcs_ctlg.h"
#include "access/cstore_vector.h"
#include "access/imcs/imcs_insert.h"
#include "access/imcs/imcu.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "utils/dynahash.h"
#include "postmaster/bgworker.h"
#include "funcapi.h"
#include "libpq/libpq-int.h"
#include "utils/guc_storage.h"
#ifdef ENABLE_HTAP
#include "storage/htap/htap_modify.h"
#endif
#include "pgxc/execRemote.h"
#include "libpq/libpq.h"
#include "catalog/toasting.h"
#include "catalog/cstore_ctlg.h"
#include "parser/parse_relation.h"

#define IMCS_HASHTABLE_MAX_SIZE 100000

int compareint2(const void *va, const void *vb)
{
    int2 a = *((const int2 *)va);
    int2 b = *((const int2 *)vb);

    if (a == b) {
        return 0;
    }
    return (a > b) ? 1 : -1;
}

/*
 * InitImcsHashTbl
 */
void InitImcsHashTbl()
{
    HASHCTL ctl;
    errno_t rc;

    g_instance.stat_cxt.IMCSInfoHashtableContext =
        AllocSetContextCreate(g_instance.instance_context, "IMCSInfoHashtableContext", ALLOCSET_SMALL_MINSIZE,
        ALLOCSET_SMALL_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE, SHARED_CONTEXT);

    rc = memset_s(&ctl, sizeof(ctl), 0, sizeof(ctl));
    securec_check_c(rc, "\0", "\0");

    ctl.hcxt = g_instance.stat_cxt.IMCSInfoHashtableContext;
    ctl.keysize = sizeof(Oid);
    ctl.entrysize = sizeof(IMCSHashEntry);
    ctl.hash = tag_hash;

    g_instance.stat_cxt.imcsHashTable = hash_create("IMCSOID to gs_imcs_info map", IMCS_HASHTABLE_MAX_SIZE, &ctl,
        HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT | HASH_EXTERN_CONTEXT);
    g_instance.stat_cxt.imcsHashTableLock = LWLockAssign(LWTRANCHE_IMCS_INFO_HASHTABLE);

    pg_atomic_init_u32(&IMCU_CACHE->m_imcs_cnt, 0);
}

void AddToImcHashTbl(Oid rel_oid, Oid imcs_oid, bool is_partition, int2vector *imcskey, int nattrs,
    XLogRecPtr current_lsn)
{
    MemoryContext oldcontext = NULL;
    IMCSHashEntry *result_found;

    oldcontext = MemoryContextSwitchTo(g_instance.stat_cxt.imcsHashTable->hcxt);
    LWLockAcquire(g_instance.stat_cxt.imcsHashTableLock, LW_EXCLUSIVE);

    result_found = (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &(rel_oid), HASH_ENTER, NULL);
    errno_t rc = EOK;
    result_found->imcsoid = imcs_oid;
    result_found->imcsispart = is_partition;
    result_found->current_lsn = current_lsn;
    result_found->imcsstatus = (char *)palloc0(sizeof(char) * NAMEDATALEN);
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
        rc = snprintf_s(result_found->imcsstatus, NAMEDATALEN, NAMEDATALEN - 1, IMCS_STATUS_ONSTANDBY);
    } else {
        rc = snprintf_s(result_found->imcsstatus, NAMEDATALEN, NAMEDATALEN - 1, IMCS_STATUS_INITIAL);
    }
    securec_check_ss(rc, "", "");
    result_found->imcsIsAvailable = false;
    result_found->colsIsAvailable = NIL;

    if (nattrs > 0) {
        result_found->imcsnattr = nattrs;
        result_found->imcskey = (int2vector *)palloc0(Int2VectorSize(nattrs));
        rc = memcpy_s(result_found->imcskey, Int2VectorSize(nattrs), imcskey, Int2VectorSize(nattrs));
        securec_check(rc, "\0", "\0");
    } else {
        result_found->imcsnattr = 0;
        result_found->imcskey = NULL;
    }

    LWLockRelease(g_instance.stat_cxt.imcsHashTableLock);
    (void)MemoryContextSwitchTo(oldcontext);
}

void RemoveFromImcHashTbl(Oid rel_oid)
{
    MemoryContext oldcontext = NULL;
    oldcontext = MemoryContextSwitchTo(g_instance.stat_cxt.imcsHashTable->hcxt);
    LWLockAcquire(g_instance.stat_cxt.imcsHashTableLock, LW_EXCLUSIVE);
    (void)hash_search(g_instance.stat_cxt.imcsHashTable, &rel_oid, HASH_REMOVE, NULL);
    LWLockRelease(g_instance.stat_cxt.imcsHashTableLock);
    (void)MemoryContextSwitchTo(oldcontext);
}

IMCSHashEntry *SearchImcHashTbl(Oid rel_oid)
{
    MemoryContext oldcontext = NULL;
    IMCSHashEntry *result_found;
    bool found;
    Form_gs_imcs gsimcs_form;

    oldcontext = MemoryContextSwitchTo(g_instance.stat_cxt.imcsHashTable->hcxt);
    LWLockAcquire(g_instance.stat_cxt.imcsHashTableLock, LW_EXCLUSIVE);
    result_found = (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &rel_oid, HASH_ENTER, &found);
    if (!found) {
        gsimcs_form = search_records_from_gs_imcs(rel_oid);
        result_found->imcsoid = gsimcs_form->imcsoid;
        result_found->imcsispart = gsimcs_form->imcsispart;
        result_found->imcsstatus = IMCS_STATUS_INITIAL;
        result_found->imcsIsAvailable = false;
        result_found->colsIsAvailable = NIL;
    }

    LWLockRelease(g_instance.stat_cxt.imcsHashTableLock);
    (void)MemoryContextSwitchTo(oldcontext);
    return result_found;
}

void UpdateImcHashTbl(Oid rel_oid, int natts, int2 *imcs_key, int imcs_natts, bool imcsstatus_judge)
{
    IMCSHashEntry *result_found = NULL;
    MemoryContext oldcontext = NULL;
    bool found = false;
    oldcontext = MemoryContextSwitchTo(g_instance.stat_cxt.imcsHashTable->hcxt);
    bool *key = (bool *)palloc0((natts + 1) * sizeof(bool));
    qsort((void *)imcs_key, imcs_natts, sizeof(int2), compareint2);

    int nkeys = imcs_natts;
    int current_key = 0;
    int col_available_num = 0;
    for (int i = 1; i < natts + 1; i++) {
        key[i] = false;
        if (current_key < nkeys && i == imcs_key[current_key]) {
            col_available_num++;
            key[i] = true;
            current_key++;
        }
    }

    LWLockAcquire(g_instance.stat_cxt.imcsHashTableLock, LW_EXCLUSIVE);
    result_found =
        (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &(rel_oid), HASH_ENTER, &found);

    result_found->imcsIsAvailable = false;
    key[0] = false;
    if (col_available_num > 0) {
        result_found->imcsIsAvailable = true;
        result_found->imcsstatus = IMCS_STATUS_COMPLETED;
        if (!imcsstatus_judge) {
            result_found->imcsstatus = IMCS_STATUS_INITIAL;
        }
        IMCSDesc *imcs_desc = IMCU_CACHE->GetImcsDesc(rel_oid);
        if (imcs_desc != NULL && imcs_desc->cu_num_in_disk > 0) {
            result_found->imcsstatus = IMCS_STATUS_STORED;
        }
        key[0] = true;
    }

    result_found->colsIsAvailable = NIL;
    for (int i = 0; i < natts + 1; i++) {
        result_found->colsIsAvailable = lappend(result_found->colsIsAvailable, &key[i]);
    }

    LWLockRelease(g_instance.stat_cxt.imcsHashTableLock);
    (void)MemoryContextSwitchTo(oldcontext);
}

void UpdateImcHashTblStatus(Oid rel_oid, char *imcs_status)
{
    IMCSHashEntry *result_found = NULL;
    MemoryContext oldcontext = NULL;
    bool found = false;
    oldcontext = MemoryContextSwitchTo(g_instance.stat_cxt.imcsHashTable->hcxt);
    LWLockAcquire(g_instance.stat_cxt.imcsHashTableLock, LW_EXCLUSIVE);
    result_found =
        (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &(rel_oid), HASH_ENTER, &found);
    errno_t rc = EOK;
    result_found->imcsstatus = (char *)palloc0(sizeof(char) * NAMEDATALEN);
    rc = snprintf_s(result_found->imcsstatus, NAMEDATALEN, NAMEDATALEN - 1, imcs_status);
    securec_check_ss(rc, "", "");
    result_found->imcsIsAvailable = false;
    LWLockRelease(g_instance.stat_cxt.imcsHashTableLock);
    (void)MemoryContextSwitchTo(oldcontext);
}

void AddSlotToImcHashTblInfoOnStandby(Oid rel_oid, TupleDesc tts_tupleDescriptor)
{
    if (t_thrd.postmaster_cxt.HaShmData->current_mode != STANDBY_MODE) {
        return;
    }
    IMCSHashEntry *result_found = NULL;
    MemoryContext oldcontext = NULL;
    bool found = false;
    oldcontext = MemoryContextSwitchTo(g_instance.stat_cxt.imcsHashTable->hcxt);
    LWLockAcquire(g_instance.stat_cxt.imcsHashTableLock, LW_EXCLUSIVE);
    result_found =
        (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &(rel_oid), HASH_FIND, &found);
    TupleTableSlot *slot = MakeTupleTableSlot(false, TableAmHeap);
    ExecSetSlotDescriptor(slot, tts_tupleDescriptor);
    slot->tts_flags = 0;
    slot->tts_buffer = InvalidBuffer;
    result_found->slot = slot;
    LWLockRelease(g_instance.stat_cxt.imcsHashTableLock);
    (void)MemoryContextSwitchTo(oldcontext);
}

void init_parallel_imcu_scan(TableScanDesc sscan, PopulateShared *bulkload_shared)
{
    HeapScanDesc scan = (HeapScanDesc)sscan;

    if (!scan || scan->rs_base.rs_nblocks == 0) {
        return;
    }

    int cur_smp_id;
    LWLockAcquire(bulkload_shared->parallel_populate_lock, LW_EXCLUSIVE);
    cur_smp_id = bulkload_shared->cur_smp_id;
    bulkload_shared->cur_smp_id++;
    LWLockRelease(bulkload_shared->parallel_populate_lock);

    scan->rs_base.rs_startblock = cur_smp_id * bulkload_shared->parallel_populate_gap;

    if ((cur_smp_id + 1) * bulkload_shared->parallel_populate_gap > bulkload_shared->total_pages) {
        scan->rs_base.rs_nblocks = bulkload_shared->total_pages;
    } else {
        scan->rs_base.rs_nblocks = (cur_smp_id + 1) * bulkload_shared->parallel_populate_gap;
    }
}

static void _init_imcu_parallel_main(const BgWorkerContext *bwc)
{
    errno_t rc = EOK;
    PopulateShared *bulkload_shared = (PopulateShared *)bwc->bgshared;
    Relation rel;
    bool is_part = (bulkload_shared->rel->grandparentId != InvalidOid || bulkload_shared->rel->parentId != InvalidOid);
    if (is_part) {
        Oid baseOid = bulkload_shared->rel->grandparentId == InvalidOid ? bulkload_shared->rel->parentId :
                                                                          bulkload_shared->rel->grandparentId;
        Relation baseRelation = heap_open(baseOid, NoLock);
        LOCKMODE lock = t_thrd.postmaster_cxt.HaShmData->current_mode == STANDBY_MODE ? NoLock : ShareLock;
        Partition part = partitionOpen(baseRelation, bulkload_shared->heaprelid, lock);
        rel = partitionGetRelation(baseRelation, part);
        partitionClose(baseRelation, part, NoLock);
        heap_close(baseRelation, NoLock);
    } else {
        rel = heap_open(bulkload_shared->heaprelid, NoLock);
    }

    IMCSBulkloadRows *batchRowsPtr =
        New(CurrentMemoryContext)IMCSBulkloadRows(bulkload_shared->tupdesc, IMCS_MAX_ROWS_SIZE, true);

    IMCSInsert *imcsInsert = New(CurrentMemoryContext)
        IMCSInsert(rel, bulkload_shared->result_rel_info, false, NULL, NULL, bulkload_shared->tupdesc);

    if (!bulkload_shared->need_populate_ctid) {
        batchRowsPtr->m_attr_num -= 1;
    }
    TableScanDesc scan;
    scan = tableam_scan_begin(rel, SnapshotNow, 0, NULL);
    init_parallel_imcu_scan(scan, bulkload_shared);

    /* values with ctid */
    Datum *values = (Datum *)palloc0((rel->rd_att->natts + 1) * sizeof(Datum));
    bool *nulls = (bool *)palloc0((rel->rd_att->natts + 1) * sizeof(bool));

    HeapTuple heap_tuple;
    int tupleCount = 0;

    while ((heap_tuple = (HeapTuple)tableam_scan_gettuple_for_parallel_populate(scan)) != NULL) {
        tableam_tops_deform_tuple(heap_tuple, rel->rd_att, values, nulls);

        CTID ctid;
        rc = memcpy_s(&ctid.item, sizeof(CTID), &heap_tuple->t_self, sizeof(ItemPointerData));
        securec_check(rc, "\0", "\0");

        ctid.reserved = 0;

        rc = memcpy_s(&values[rel->rd_att->natts], sizeof(Datum), &ctid, sizeof(CTID));
        securec_check(rc, "\0", "\0");

        nulls[rel->rd_att->natts] = false;
        batchRowsPtr->append_one_tuple(values, nulls, NULL, rel->rd_att->natts, bulkload_shared->tupdesc,
            bulkload_shared->imcs_key_need_populate, bulkload_shared->imcs_need_populate_nkeys);
        tupleCount++;
        if (tupleCount >= IMCS_MAX_ROWS_SIZE) {
            LWLockAcquire(bulkload_shared->parallel_populate_lock, LW_EXCLUSIVE);
            imcsInsert->m_cur_cuid = bulkload_shared->m_cur_cuid_parallel;
            bulkload_shared->m_cur_cuid_parallel = bulkload_shared->m_cur_cuid_parallel + 1;
            LWLockRelease(bulkload_shared->parallel_populate_lock);

            imcsInsert->BatchInsertCommon(batchRowsPtr, 0);
            batchRowsPtr->reset(true);
            tupleCount = 0;
        }
    }

    if (batchRowsPtr->m_rows_curnum > 0) {
        SpinLockAcquire(&imcsInsert->populate_mutex);
        imcsInsert->m_cur_cuid = bulkload_shared->m_cur_cuid_parallel;
        bulkload_shared->m_cur_cuid_parallel = bulkload_shared->m_cur_cuid_parallel + 1;
        SpinLockRelease(&imcsInsert->populate_mutex);

        imcsInsert->BatchInsertCommon(batchRowsPtr, 0);
        batchRowsPtr->reset(true);
    }

    imcsInsert->SetEndFlag();
    tableam_scan_end(scan);
    pfree(values);
    pfree(nulls);

    DELETE_EX(batchRowsPtr);
    DELETE_EX(imcsInsert);

    if (is_part) {
        releaseDummyRelation(&rel);
    } else {
        heap_close(rel, NoLock);
    }
}

static void _init_imcu_parallel_cleanup(const BgWorkerContext *bwc)
{
    // todo
}

// wait for all scan workers finished
static double _init_imcu_parallel_heapscan(PopulateState *state, bool *brokenhotchain, double **alltuples)
{
    PopulateLeader *leader = state->leader;
    PopulateShared *bulkload_shared = leader->bulkload_shared;
    double reltuples;

    BgworkerListWaitFinish(&leader->nparticipants);
    /* no need to lock due to all bgworkers were terminated */
    pg_memory_barrier();

    reltuples = bulkload_shared->reltuples;

    return reltuples;
}

void initImcuOnRel(Relation rel, int2 *imcskeys, int imcsnattr)
{
    IMCSBulkloadRows *batchRowsPtr = NULL;
    IMCSInsert *imcsInsert = NULL;
    errno_t rc = EOK;
    // Step 1: Determine whether the table needs to be populated and which columns need to be populated
    int imcs_natts = imcsnattr;
    int2 *imcs_key = (int2 *)palloc0(imcs_natts * sizeof(int2));
    rc = memcpy_s(imcs_key, imcs_natts * sizeof(int2), imcskeys, imcs_natts * sizeof(int2));
    securec_check(rc, "\0", "\0");

    int2 *imcs_key_need_populate;
    int imcs_need_populate_nkeys = 0;
    qsort((void *)imcs_key, imcs_natts, sizeof(int2), compareint2);

    bool found = false;
    // whether we need to add ctid column into imcu
    bool need_populate_ctid = false;

    imcs_key_need_populate = imcs_key;
    imcs_need_populate_nkeys = imcs_natts;
    need_populate_ctid = true;

    // Step 2: Create a tupdesc that contains ctid.
    TupleDesc tupdesc;
    tupdesc = CreateTemplateTupleDesc(imcs_need_populate_nkeys + 1, false);
    TupleDescInitEntry(tupdesc, imcs_need_populate_nkeys + 1, "ctid", TIDOID, -1, 0);
    tupdesc->attrs[imcs_need_populate_nkeys].attnum = VirtualCtidColID;
    tupdesc->attrs[imcs_need_populate_nkeys].attlen = sizeof(CTID);
    tupdesc->attrs[imcs_need_populate_nkeys].attnotnull = false;
    tupdesc->attrs[imcs_need_populate_nkeys].atthasdef = false;

    // copy tupledesc base info
    tupdesc->tdtypeid = rel->rd_att->tdtypeid;
    tupdesc->tdtypmod = rel->rd_att->tdtypmod;
    tupdesc->tdisredistable = rel->rd_att->tdisredistable;

    /* copy attr info */
    for (int i = 0; i < imcs_need_populate_nkeys; i++) {
        AttrNumber attnum;
        attnum = imcs_key_need_populate[i];
        rc = memcpy_s(&tupdesc->attrs[i], ATTRIBUTE_FIXED_PART_SIZE, &rel->rd_att->attrs[attnum - 1],
            ATTRIBUTE_FIXED_PART_SIZE);
        securec_check(rc, "\0", "\0");
        tupdesc->attrs[i].attnotnull = false;
        tupdesc->attrs[i].atthasdef = false;
    }

    ResultRelInfo *result_rel_info = makeNode(ResultRelInfo);
    InitResultRelInfo(result_rel_info, rel, 1, 0);

    // Step 3: Initialize batchRowPtr and imcsInsert
    batchRowsPtr = New(CurrentMemoryContext)IMCSBulkloadRows(tupdesc, IMCS_MAX_ROWS_SIZE, true);
    imcsInsert = New(CurrentMemoryContext)IMCSInsert(rel, result_rel_info, false, NULL, NULL, tupdesc);
    if (!need_populate_ctid) {
        batchRowsPtr->m_attr_num -= 1;
    }

    Datum *values = (Datum *)palloc0((rel->rd_att->natts + 1) * sizeof(Datum));
    bool *nulls = (bool *)palloc0((rel->rd_att->natts + 1) * sizeof(bool));

    TableScanDesc scan = tableam_scan_begin(rel, SnapshotNow, 0, NULL);
    int tupleCount = 0;
    HeapScanDesc heapScanDesc = (HeapScanDesc)scan;
    AddSlotToImcHashTblInfoOnStandby(RelationGetRelid(rel), heapScanDesc->rs_tupdesc);
    IMCU_CACHE->SaveImcsDesc(RelationGetRelid(rel));
    pg_atomic_add_fetch_u32(&IMCU_CACHE->m_imcs_cnt, 1);

    HeapTuple heap_tuple;
    while ((heap_tuple = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection)) != NULL) {
        tableam_tops_deform_tuple(heap_tuple, rel->rd_att, values, nulls);

        CTID ctid;
        rc = memcpy_s(&ctid.item, sizeof(CTID), &heap_tuple->t_self, sizeof(ItemPointerData));
        securec_check(rc, "\0", "\0");

        ctid.reserved = 0;

        rc = memcpy_s(&values[rel->rd_att->natts], sizeof(Datum), &ctid, sizeof(CTID));
        securec_check(rc, "\0", "\0");

        nulls[rel->rd_att->natts] = false;
        batchRowsPtr->append_one_tuple(values, nulls, NULL, rel->rd_att->natts, tupdesc, imcs_key_need_populate,
            imcs_need_populate_nkeys);
        tupleCount++;
        if (tupleCount >= IMCS_MAX_ROWS_SIZE) {
            imcsInsert->BatchInsertCommon(batchRowsPtr, 0);
            batchRowsPtr->reset(true);
            tupleCount = 0;
        }
    }

    if (batchRowsPtr->m_rows_curnum > 0) {
        imcsInsert->BatchInsertCommon(batchRowsPtr, 0);
        batchRowsPtr->reset(true);
    }

    imcsInsert->SetEndFlag();
    tableam_scan_end(scan);
    pfree(values);
    pfree(nulls);

    DELETE_EX(batchRowsPtr);
    DELETE_EX(imcsInsert);
    FreeTupleDesc(tupdesc);
    if (result_rel_info != NULL) {
        ExecCloseIndices(result_rel_info);
        pfree(result_rel_info);
    }
}

void InitImcuParallel(Relation rel, int2 *imcskeys, int imcsnattr)
{
    PopulateState state;
    state.heap_rel = rel;
    state.leader = NULL;

    // step 1 start n workers
    PopulateShared *bulkload_shared;
    PopulateLeader *leader = (PopulateLeader *)palloc0(sizeof(PopulateLeader));

    /* Launch workers, saving status for leader/caller */

    bulkload_shared = (PopulateShared *)MemoryContextAllocZero(INSTANCE_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE),
        sizeof(PopulateShared));
    bulkload_shared->heaprelid = RelationGetRelid(rel);
    SpinLockInit(&bulkload_shared->mutex);
    bulkload_shared->reltuples = 0.0;

    HeapParallelscanInitialize(&bulkload_shared->heapdesc, rel);

    errno_t rc = EOK;

    int imcs_natts = imcsnattr;
    int2 *imcs_key = (int2 *)palloc0(imcs_natts * sizeof(int2));
    rc = memcpy_s(imcs_key, imcs_natts * sizeof(int2), imcskeys, imcs_natts * sizeof(int2));
    securec_check(rc, "\0", "\0");

    int2 *imcs_key_need_populate;
    int imcs_need_populate_nkeys = 0;
    qsort((void *)imcs_key, imcs_natts, sizeof(int2), compareint2);

    IMCSHashEntry *result_found = NULL;
    bool found = false;
    bool need_populate_ctid = false;
    result_found = (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &(RelationGetRelid(rel)),
        HASH_FIND, &found);
    imcs_key_need_populate = imcs_key;
    imcs_need_populate_nkeys = imcs_natts;
    need_populate_ctid = true;

    TableScanDesc scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
    HeapScanDesc heapScanDesc = (HeapScanDesc)scan;
    AddSlotToImcHashTblInfoOnStandby(RelationGetRelid(rel), heapScanDesc->rs_tupdesc);
    IMCU_CACHE->SaveImcsDesc(RelationGetRelid(rel));
    pg_atomic_add_fetch_u32(&IMCU_CACHE->m_imcs_cnt, 1);

    bulkload_shared->imcs_key_need_populate = imcs_key_need_populate;
    bulkload_shared->imcs_need_populate_nkeys = imcs_need_populate_nkeys;

    TupleDesc tupdesc;
    tupdesc = CreateTemplateTupleDesc(imcs_need_populate_nkeys + 1, false);
    TupleDescInitEntry(tupdesc, imcs_need_populate_nkeys + 1, "ctid", TIDOID, -1, 0);
    tupdesc->attrs[imcs_need_populate_nkeys].attnum = VirtualCtidColID;
    tupdesc->attrs[imcs_need_populate_nkeys].attlen = sizeof(CTID);
    tupdesc->attrs[imcs_need_populate_nkeys].attnotnull = false;
    tupdesc->attrs[imcs_need_populate_nkeys].atthasdef = false;

    /* copy tupledesc base info */
    tupdesc->tdtypeid = rel->rd_att->tdtypeid;
    tupdesc->tdtypmod = rel->rd_att->tdtypmod;
    tupdesc->tdisredistable = rel->rd_att->tdisredistable;

    /* copy attr info */
    for (int i = 0; i < imcs_need_populate_nkeys; i++) {
        AttrNumber attnum;
        attnum = imcs_key_need_populate[i];
        rc = memcpy_s(&tupdesc->attrs[i], ATTRIBUTE_FIXED_PART_SIZE, &rel->rd_att->attrs[attnum - 1],
            ATTRIBUTE_FIXED_PART_SIZE);
        securec_check(rc, "\0", "\0");
        tupdesc->attrs[i].attnotnull = false;
        tupdesc->attrs[i].atthasdef = false;
    }

    ResultRelInfo *result_rel_info = makeNode(ResultRelInfo);
    InitResultRelInfo(result_rel_info, rel, 1, 0);

    bulkload_shared->tupdesc = tupdesc;
    bulkload_shared->result_rel_info = result_rel_info;
    bulkload_shared->m_cur_cuid_parallel = 0;
    bulkload_shared->rel = rel;
    bulkload_shared->need_populate_ctid = need_populate_ctid;
    bulkload_shared->cur_smp_id = 0;

    int nworkers;

    nworkers = (scan->rs_nblocks / IMCSTORED_NUM_BLOCKS_FOR_SCAN) + 1;
    if (nworkers > 10) {
        nworkers = 10;
    }

    bulkload_shared->parallel_populate_gap =
        (scan->rs_nblocks % nworkers) ? (scan->rs_nblocks / nworkers) + 1 : (scan->rs_nblocks / nworkers);
    bulkload_shared->total_pages = scan->rs_nblocks;
    bulkload_shared->parallel_populate_lock = LWLockAssign(LWTRANCHE_POPULATE_SHARED);
    heap_endscan(scan);
    leader->nparticipants =
        LaunchBackgroundWorkers(nworkers, bulkload_shared, _init_imcu_parallel_main, _init_imcu_parallel_cleanup);

    /* if all workers start failed, will series populate todo */
    if (leader->nparticipants == 0) {
        pfree_ext(bulkload_shared);
        pfree_ext(leader);
        FreeTupleDesc(tupdesc);
        return;
    }

    // step 2 wait workers finish and update metadata
    double reltuples = 0;
    leader->bulkload_shared = bulkload_shared;
    state.leader = leader;
    reltuples = _init_imcu_parallel_heapscan(&state, NULL, NULL);
    bulkload_shared = state.leader->bulkload_shared;

    // step3 end worker
    if (state.leader) {
        BgworkerListSyncQuit();
    }
}

void DropImcs(Oid rel_oid, RelFileNode rd_node)
{
    CleanImcsHashTbl(rel_oid);
    IMCSDesc *imcs_desc = IMCU_CACHE->GetImcsDesc(rel_oid);
    if (imcs_desc != NULL) {
        LWLockAcquire(imcs_desc->imcu_desc_latch, LW_EXCLUSIVE);
        IMCU_CACHE->DropRelationCUCache(rd_node);
        LWLockRelease(imcs_desc->imcu_desc_latch);
        if (imcs_desc->cu_num_in_disk != 0) {
            IMCU_CACHE->DropRelationImcuFile(rd_node, imcs_desc);
        }
        IMCU_CACHE->DropImcsDesc(rel_oid);
    }

    RemoveFromImcHashTbl(rel_oid);
    CommandCounterIncrement();
}

void CleanImcsHashTbl(Oid rel_oid)
{
    MemoryContext oldcontext = NULL;
    IMCSHashEntry *result_found = NULL;
    oldcontext = MemoryContextSwitchTo(g_instance.stat_cxt.imcsHashTable->hcxt);
    LWLockAcquire(g_instance.stat_cxt.imcsHashTableLock, LW_EXCLUSIVE);
    bool found = false;
    result_found =
        (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &(rel_oid), HASH_ENTER, &found);
    if (!found || result_found->imcsIsAvailable == false) {
        result_found->colsIsAvailable = NIL;
        LWLockRelease(g_instance.stat_cxt.imcsHashTableLock);
        (void)MemoryContextSwitchTo(oldcontext);
        return;
    }

    result_found->imcsIsAvailable = false;
    ListCell *lc;
    foreach (lc, result_found->colsIsAvailable) {
        bool *col_is_available = (bool *)lfirst(lc);
        *col_is_available = false;
    }
    LWLockRelease(g_instance.stat_cxt.imcsHashTableLock);
    (void)MemoryContextSwitchTo(oldcontext);
}

void RemoveImcstore(Relation rel, std::unordered_map<Oid, RelFileNode> *oid_del_info, bool isPartitioned)
{
    if (oid_del_info && IMCU_CACHE->m_reinit_imcs_set.count(RelationGetRelid(rel)) == 0) {
        (*oid_del_info)[RelationGetRelid(rel)] = rel->rd_node;
        return;
    }

    if (isPartitioned) {
        Oid imcs_oid = InvalidOid;
        LOCKMODE lock = t_thrd.postmaster_cxt.HaShmData->current_mode == STANDBY_MODE ? NoLock : AccessExclusiveLock;
        if (RelationIsSubPartitioned(rel)) {
            Oid part_oid;
            ListCell *cell = NULL;
            List *part_oid_list = RelationGetSubPartitionOidList(rel, lock);
            foreach (cell, part_oid_list) {
                part_oid = lfirst_oid(cell);
                imcs_oid = search_table_from_gs_imcs(part_oid);
                if (OidIsValid(imcs_oid)) {
                    Partition part = partitionOpen(rel, part_oid, lock);
                    Relation part_rel = partitionGetRelation(rel, part);
                    DropImcs(part_oid, part_rel->rd_node);
                    releaseDummyRelation(&part_rel);
                    partitionClose(rel, part, NoLock);
                }
            }
            if (part_oid_list != NULL) {
                releasePartitionOidList(&part_oid_list);
            }
        } else {
            Oid part_oid;
            ListCell *cell = NULL;
            List *part_oid_list = relationGetPartitionOidList(rel);
            foreach (cell, part_oid_list) {
                part_oid = lfirst_oid(cell);
                imcs_oid = search_table_from_gs_imcs(part_oid);
                if (OidIsValid(imcs_oid)) {
                    Partition part = partitionOpen(rel, part_oid, lock);
                    Relation part_rel = partitionGetRelation(rel, part);
                    DropImcs(part_oid, part_rel->rd_node);
                    releaseDummyRelation(&part_rel);
                    partitionClose(rel, part, NoLock);
                }
            }
            if (part_oid_list != NULL) {
                releasePartitionOidList(&part_oid_list);
            }
        }
    } else {
        DropImcs(RelationGetRelid(rel), rel->rd_node);
    }
    return;
}

void TruncateImcs(Oid rel_oid, RelFileNode rd_node)
{
    IMCSDesc *imcs_desc = IMCU_CACHE->GetImcsDesc(rel_oid);
    if (imcs_desc != NULL) {
        LWLockAcquire(imcs_desc->imcu_desc_latch, LW_EXCLUSIVE);
        IMCU_CACHE->DropRelationCUCache(rd_node);
        LWLockRelease(imcs_desc->imcu_desc_latch);
        if (imcs_desc->cu_num_in_disk != 0) {
            IMCU_CACHE->DropRelationImcuFile(rd_node, imcs_desc);
        }
        IMCU_CACHE->DropImcsDesc(rel_oid);
        IMCU_CACHE->SaveImcsDesc(rel_oid);
    }
}

void TruncateImcstore(Relation rel, std::unordered_map<Oid, RelFileNode> *oid_del_info)
{
    if (oid_del_info) {
        (*oid_del_info)[RelationGetRelid(rel)] = rel->rd_node;
        return;
    }
    TruncateImcs(RelationGetRelid(rel), rel->rd_node);
}

int GetImcsAttsKeys(Oid rel_oid, int2 **imcs_key)
{
    ServerMode current_mode = t_thrd.postmaster_cxt.HaShmData->current_mode;
    int imcs_natts = 0;
    errno_t rc = EOK;
    if (current_mode == STANDBY_MODE) {
        bool found = false;
        IMCSHashEntry *result_found = NULL;
        result_found =
            (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &rel_oid, HASH_FIND, &found);
        if (!found) {
            return imcs_natts;
        }
        imcs_natts = result_found->imcsnattr;
        *imcs_key = (int2 *)palloc0(imcs_natts * sizeof(int2));
        rc = memcpy_s(*imcs_key, imcs_natts * sizeof(int2), result_found->imcskey->values, imcs_natts * sizeof(int2));
        securec_check(rc, "\0", "\0");
        return imcs_natts;
    } else {
        Relation imcs_rel = NULL;
        TableScanDesc scan;
        Form_gs_imcs gsimcs_form;
        HeapTuple heap_tup = NULL;
        TupleDesc imcstup_desc;

        /* search rel_oid in gs_imcs catalog */
        imcs_rel = heap_open(ImcsRelationId, RowExclusiveLock);
        imcstup_desc = RelationGetDescr(imcs_rel);
        scan = tableam_scan_begin(imcs_rel, SnapshotNow, 0, NULL);
        heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
        while (heap_tup != NULL) {
            gsimcs_form = (Form_gs_imcs)GETSTRUCT(heap_tup);
            if (gsimcs_form->reloid == rel_oid) {
                /* found tuple */
                break;
            }
            heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
        }

        if (!HeapTupleIsValid(heap_tup)) {
            heap_endscan(scan);
            heap_close(imcs_rel, NoLock);
            return imcs_natts;
        }

        imcs_natts = gsimcs_form->imcsnattr;
        *imcs_key = (int2 *)palloc0(imcs_natts * sizeof(int2));
        rc = memcpy_s(*imcs_key, imcs_natts * sizeof(int2), gsimcs_form->imcskey.values, imcs_natts * sizeof(int2));
        securec_check(rc, "\0", "\0");
        heap_endscan(scan);
        heap_close(imcs_rel, NoLock);
        return imcs_natts;
    }
}

void InitImcstore(Oid rel_oid, Relation rel, bool isPartitioned)
{
    if (isPartitioned) {
        Oid imcs_oid = InvalidOid;
        LOCKMODE lock = t_thrd.postmaster_cxt.HaShmData->current_mode == STANDBY_MODE ? NoLock : ShareLock;
        if (RelationIsSubPartitioned(rel)) {
            Oid part_oid;
            ListCell *cell = NULL;
            List *part_oid_list = RelationGetSubPartitionOidList(rel, lock);
            foreach (cell, part_oid_list) {
                part_oid = lfirst_oid(cell);
                imcs_oid = search_table_from_gs_imcs(part_oid);
                if (OidIsValid(imcs_oid)) {
                    Partition part = partitionOpen(rel, part_oid, lock);
                    Relation part_rel = partitionGetRelation(rel, part);
                    InitImcstore(part_oid, part_rel, false);
                    releaseDummyRelation(&part_rel);
                    partitionClose(rel, part, NoLock);
                }
            }
        } else {
            Oid part_oid;
            ListCell *cell = NULL;
            List *part_oid_list = relationGetPartitionOidList(rel);
            foreach (cell, part_oid_list) {
                part_oid = lfirst_oid(cell);
                imcs_oid = search_table_from_gs_imcs(part_oid);
                if (OidIsValid(imcs_oid)) {
                    Partition part = partitionOpen(rel, part_oid, lock);
                    Relation part_rel = partitionGetRelation(rel, part);
                    InitImcstore(part_oid, part_rel, false);
                    releaseDummyRelation(&part_rel);
                    partitionClose(rel, part, NoLock);
                }
            }
        }
    } else {
        int2 *imcskey;
        int col_num = 0;
        col_num = GetImcsAttsKeys(rel_oid, &imcskey);

        if (col_num == 0) {
            return;
        }
        UpdateImcHashTblStatus(rel_oid, IMCS_STATUS_STARTING);
        if ((u_sess->attr.attr_common.enable_parallel_init_imcs)) {
            InitImcuParallel(rel, imcskey, col_num);
        } else {
            initImcuOnRel(rel, imcskey, col_num);
        }
        UpdateImcHashTbl(rel_oid, rel->rd_att->natts, imcskey, col_num);
    }
}

void CreateImcDeltaTable(Oid rel_id)
{
#ifdef ENABLE_HTAP
    HTAPCreateTable(rel_id, rel_id);
#endif
}

void CreatePartitionImcDeltaTable(Oid rel_id, Oid part_id)
{
#ifdef ENABLE_HTAP
    HTAPCreateTable(rel_id, part_id);
#endif
}

void CreateImcDeltaTableForPartitions(Oid rel_id, Relation rel)
{
    Oid imcs_oid = InvalidOid;
    LOCKMODE lock = t_thrd.postmaster_cxt.HaShmData->current_mode == STANDBY_MODE ? NoLock : ShareLock;
    if (RelationIsSubPartitioned(rel)) {
        Oid part_oid;
        ListCell *cell = NULL;
        List *part_oid_list = RelationGetSubPartitionOidList(rel, lock);
        foreach (cell, part_oid_list) {
            part_oid = lfirst_oid(cell);
            imcs_oid = search_table_from_gs_imcs(part_oid);
            if (OidIsValid(imcs_oid)) {
                Partition part = partitionOpen(rel, part_oid, lock);
                Relation part_rel = partitionGetRelation(rel, part);
                CreatePartitionImcDeltaTable(rel_id, part_oid);
                releaseDummyRelation(&part_rel);
                partitionClose(rel, part, NoLock);
            }
        }
    } else {
        Oid part_oid;
        ListCell *cell = NULL;
        List *part_oid_list = relationGetPartitionOidList(rel);
        foreach (cell, part_oid_list) {
            part_oid = lfirst_oid(cell);
            imcs_oid = search_table_from_gs_imcs(part_oid);
            if (OidIsValid(imcs_oid)) {
                Partition part = partitionOpen(rel, part_oid, lock);
                Relation part_rel = partitionGetRelation(rel, part);
                CreatePartitionImcDeltaTable(rel_id, part_oid);
                releaseDummyRelation(&part_rel);
                partitionClose(rel, part, NoLock);
            }
        }
    }
}

static int HandleImcsResponse(PGXCNodeHandle *conn, RemoteQueryState *combiner)
{
    char *msg = NULL;
    int msg_len;
    char msg_type;
    bool error_flag = false;

    for (;;) {
        Assert(conn->state != DN_CONNECTION_STATE_IDLE);

        /*
         * If we are in the process of shutting down, we
         * may be rolling back, and the buffer may contain other messages.
         * We want to avoid a procarray exception
         * as well as an error stack overflow.
         *
         * If not in GPC mode, should receive datanode messages but not interrupt immediately in loop while.
         */
        if (t_thrd.proc_cxt.proc_exit_inprogress && ENABLE_CN_GPC) {
            conn->state = DN_CONNECTION_STATE_ERROR_FATAL;
            ereport(DEBUG2, (errmsg(
                "DN_CONNECTION_STATE_ERROR_FATAL0 is set for connection to node %s[%u] when proc_exit_inprogress",
                conn->remoteNodeName, conn->nodeoid)));
        }

        /* don't read from from the connection if there is a fatal error */
        if (conn->state == DN_CONNECTION_STATE_ERROR_FATAL) {
            ereport(DEBUG2,
                (errmsg("handle_response0 returned with DN_CONNECTION_STATE_ERROR_FATAL for connection to node %s[%u] ",
                conn->remoteNodeName, conn->nodeoid)));
            return RESPONSE_COMPLETE;
        }

        /* No data available, read one more time or exit */
        if (!HAS_MESSAGE_BUFFERED(conn)) {
            /*
             * For FATAL error, no need to read once more, because openGauss thread(DN) will exit
             * immediately after sending error message without sending 'Z'(ready for query).
             */
            if (combiner != NULL && combiner->is_fatal_error) {
                conn->state = DN_CONNECTION_STATE_ERROR_FATAL;
                conn->combiner = NULL;

                return RESPONSE_COMPLETE;
            }

            if (error_flag) {
                /* incomplete message, if last message type is ERROR,read once more */
                if (pgxc_node_receive(1, &conn, NULL))
                    ereport(ERROR, (errcode(ERRCODE_CONNECTION_EXCEPTION),
                        errmsg("Failed to receive message from %s[%u]", conn->remoteNodeName, conn->nodeoid)));
                error_flag = false;
                continue;
            } else {
                return RESPONSE_EOF;
            }
        }
        /* no need to check conn's combiner when abort transaction */
        Assert(t_thrd.xact_cxt.bInAbortTransaction || conn->combiner == combiner || conn->combiner == NULL);

        msg_type = get_message(conn, &msg_len, &msg);

        switch (msg_type) {
            case '\0': /* Not enough data in the buffer */
                return RESPONSE_EOF;
            case 'O': /* Complete */
                conn->state = DN_CONNECTION_STATE_IDLE;
                return RESPONSE_PLANID_OK;
            case 'I': /* EmptyQuery */
            default:
                /* sync lost? */
                elog(WARNING, "Received unsupported message type: %c", msg_type);
                conn->state = DN_CONNECTION_STATE_ERROR_FATAL;
                /* stop reading */
                return RESPONSE_COMPLETE;
        }
    }
    /* never happen, but keep compiler quiet */
    return RESPONSE_EOF;
}

void InitImcsOnStandby(Oid rel_id, StringInfo input_message)
{
    errno_t rc = EOK;
    /* get column_name_list */
    int count = 0;
    rc = memcpy_s(&count, sizeof(int), pq_getmsgbytes(input_message, sizeof(int)), sizeof(int));
    securec_check(rc, "\0", "\0");

    char **column_list = NULL;
    if (count > 0) {
        column_list = (char **)palloc(count * sizeof(char *));
        for (int i = 0; i < count; i++)
            column_list[i] = (char *)pq_getmsgstring(input_message);
    }

    /* get current lsn from primary */
    XLogRecPtr current_lsn;
    rc = memcpy_s(&current_lsn, sizeof(XLogRecPtr), pq_getmsgbytes(input_message, sizeof(XLogRecPtr)),
        sizeof(XLogRecPtr));
    securec_check(rc, "\0", "\0");
    ereport(DEBUG1, (errmsg("Received current_lsn for HTAP population.")));
    pq_getmsgend(input_message);

    /* Check whether the standby node has been rolled back to the current LSN. */
    do {
        if (current_lsn <= IMCU_CACHE->m_startup_cur_lsn) {
            break;
        }
    } while (true);

    /* Make sure we are in a transaction command */
    start_xact_command();
    PushActiveSnapshot(GetTransactionSnapshot(false));
    Relation rel = heap_open(rel_id, NoLock);
    int j = 0;
    int2vector *imcskey = buildint2vector(NULL, count);
    for (int cnt = 0; cnt < count; cnt++) {
        AttrNumber attnum = get_attnum(rel_id, column_list[cnt]);
        imcskey->values[j] = attnum;
        j++;
    }

    IMCSHashEntry *result_found = NULL;
    bool found = false;
    result_found = (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &rel_id, HASH_FIND, &found);
    if (result_found) {
        IMCU_CACHE->m_reinit_imcs_set.insert(rel_id);
        ClearBeforeReinitImcs(rel);
    }
    AddToImcHashTbl(rel_id, NULL, RELATION_IS_PARTITIONED(rel), imcskey, count, current_lsn);

    if (RELATION_IS_PARTITIONED(rel)) {
        Oid part_oid;
        ListCell *cell = NULL;
        List *part_oid_list = relationGetPartitionOidList(rel);
        foreach (cell, part_oid_list) {
            part_oid = lfirst_oid(cell);
            AddToImcHashTbl(part_oid, NULL, true, imcskey, count, current_lsn);

            if (RelationIsSubPartitioned(rel)) {
                Oid subpart_oid;
                ListCell *subcell = NULL;
                List *subpart_oid_list = PartOidGetSubPartitionOidList(rel, part_oid, NoLock); // subpartition oid list
                foreach (subcell, subpart_oid_list) {
                    subpart_oid = lfirst_oid(subcell);
                    AddToImcHashTbl(subpart_oid, NULL, true, imcskey, count, current_lsn);
                }
                if (subpart_oid_list != NULL) {
                    releasePartitionOidList(&subpart_oid_list);
                }
            }
        }
        if (part_oid_list != NULL) {
            releasePartitionOidList(&part_oid_list);
        }
        InitImcstore(rel_id, rel, true);
    } else {
        InitImcstore(rel_id, rel, false);
    }

    if (RELATION_IS_PARTITIONED(rel)) {
        CreateImcDeltaTableForPartitions(rel_id, rel);
    } else {
        CreateImcDeltaTable(rel_id);
    }
    heap_close(rel, NoLock);

    init_imcs_current_lsn(current_lsn);
    pq_putemptymessage('O'); /* PlanIdComplete */
    pq_flush();
    PopActiveSnapshot();
    finish_xact_command();
}

void init_imcs_current_lsn(XLogRecPtr current_lsn)
{
    LWLockAcquire(IMCU_CACHE->m_xlog_lsn_lock, LW_EXCLUSIVE);
    if (IMCU_CACHE->m_xlog_current_lsn == InvalidXLogRecPtr) {
        IMCU_CACHE->m_xlog_current_lsn = current_lsn;
    }
    LWLockRelease(IMCU_CACHE->m_xlog_lsn_lock);
}

void ClearBeforeReinitImcs(Relation rel)
{
    Oid rel_oid = RelationGetRelid(rel);
    std::unordered_map<Oid, RelFileNode> *oid_del_info = nullptr;
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
            CleanImcsHashTbl(rel_oid);
            delete_gs_imcs_table_on_rel(rel, nullptr);
            return;
    }

    if (RELATION_IS_PARTITIONED(rel)) {
        DropImcDeltaTableForPartitions(rel_oid, rel, &oid_del_info);
    } else {
        DropImcDeltaTable(rel_oid, &oid_del_info);
    }
    RemoveImcstore(rel, oid_del_info, RELATION_IS_PARTITIONED(rel));
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == NORMAL_MODE) {
        delete_gs_imcs_table_on_rel(rel, oid_del_info);
    }
    return;
}

void InitPartitionedImcstoreOnStandby(Oid rel_id, StringInfo input_message)
{
    errno_t rc = EOK;

    /* get column_name_list */
    int count = 0;
    rc = memcpy_s(&count, sizeof(int), pq_getmsgbytes(input_message, sizeof(int)), sizeof(int));
    securec_check(rc, "\0", "\0");

    char **column_list = NULL;

    if (count > 0) {
        int i;
        column_list = (char **)palloc(count * sizeof(char *));
        for (i = 0; i < count; i++)
            column_list[i] = (char *)pq_getmsgstring(input_message);
    }

    /* get current lsn from primary */
    XLogRecPtr current_lsn = InvalidXLogRecPtr;
    rc = memcpy_s(&current_lsn, sizeof(XLogRecPtr), pq_getmsgbytes(input_message, sizeof(XLogRecPtr)),
        sizeof(XLogRecPtr));
    securec_check(rc, "\0", "\0");
    ereport(DEBUG1, (errmsg("Received current_lsn for HTAP population.")));

    Oid part_id = InvalidOid;
    rc = memcpy_s(&part_id, sizeof(Oid), pq_getmsgbytes(input_message, sizeof(Oid)), sizeof(Oid));
    securec_check(rc, "\0", "\0");

    Oid subpart_id = InvalidOid;
    rc = memcpy_s(&subpart_id, sizeof(Oid), pq_getmsgbytes(input_message, sizeof(Oid)), sizeof(Oid));
    securec_check(rc, "\0", "\0");
    pq_getmsgend(input_message);

    /* Check whether the standby node has been rolled back to the current LSN. */
    do {
        if (current_lsn <= IMCU_CACHE->m_startup_cur_lsn) {
            break;
        }
    } while (true);

    /* Make sure we are in a transaction command */
    start_xact_command();
    PushActiveSnapshot(GetTransactionSnapshot(false));
    Relation rel = heap_open(rel_id, NoLock);
    int j = 0;
    int2vector *imcskey = buildint2vector(NULL, count);
    for (int cnt = 0; cnt < count; cnt++) {
        AttrNumber attnum = get_attnum(rel_id, column_list[cnt]);
        imcskey->values[j] = attnum;
        j++;
    }

    IMCSHashEntry *result_found = NULL;
    bool found = false;
    result_found = (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &rel_id, HASH_FIND, &found);
    if (!found) {
        AddToImcHashTbl(rel_id, NULL, RELATION_IS_PARTITIONED(rel), imcskey, count, current_lsn);
    }

    result_found = (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &part_id, HASH_FIND, &found);
    if (!found || result_found->imcsIsAvailable == false) {
        AddToImcHashTbl(part_id, NULL, true, imcskey, count, current_lsn);
        Partition part = partitionOpen(rel, part_id, NoLock);
        Relation part_rel = partitionGetRelation(rel, part);
        if (!RelationIsSubPartitioned(rel)) {
            InitImcstore(part_id, part_rel, false);
            CreatePartitionImcDeltaTable(rel_id, part_id);
        }
        releaseDummyRelation(&part_rel);
        partitionClose(rel, part, NoLock);
    }

    if (RelationIsSubPartitioned(rel)) {
        ListCell *cell = NULL;
        if (subpart_id != InvalidOid) {
            Partition subpart = partitionOpen(rel, subpart_id, NoLock);
            Relation subpart_rel = partitionGetRelation(rel, subpart);
            result_found =
                (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &subpart_id, HASH_FIND, &found);
            if (!found) {
                AddToImcHashTbl(subpart_id, NULL, true, imcskey, count, current_lsn);
                InitImcstore(subpart_id, subpart_rel, false);
                /* create delta table for all subpartitions */
                CreatePartitionImcDeltaTable(rel_id, subpart_id);
            }
            releaseDummyRelation(&subpart_rel);
            partitionClose(rel, subpart, NoLock);
        } else {
            List *subpart_oid_list = PartOidGetSubPartitionOidList(rel, part_id, NoLock); // subpartition oid list
            foreach (cell, subpart_oid_list) {
                Oid subpart_oid = lfirst_oid(cell);
                Partition subpart = partitionOpen(rel, subpart_oid, NoLock);
                Relation subpart_rel = partitionGetRelation(rel, subpart);
                result_found = (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &subpart_oid,
                    HASH_FIND, &found);
                if (!found) {
                    AddToImcHashTbl(subpart_oid, NULL, true, imcskey, count, current_lsn);
                    InitImcstore(subpart_oid, subpart_rel, false);
                    /* create delta table for all subpartitions */
                    CreatePartitionImcDeltaTable(rel_id, subpart_oid);
                }
                releaseDummyRelation(&subpart_rel);
                partitionClose(rel, subpart, NoLock);
            }
        }
    }
    init_imcs_current_lsn(current_lsn);

    heap_close(rel, NoLock);
    pq_putemptymessage('O'); /* PlanIdComplete */
    pq_flush();
    PopActiveSnapshot();
    finish_xact_command();
}

void RemoveImcstoreOnStandby(Oid rel_id)
{
    /* Make sure we are in a transaction command */
    start_xact_command();
    Relation rel = heap_open(rel_id, NoLock);
    std::unordered_map<Oid, RelFileNode> *oid_del_info = nullptr;
    if (RELATION_IS_PARTITIONED(rel)) {
        DropImcDeltaTableForPartitions(rel_id, rel, &oid_del_info);
    } else {
        DropImcDeltaTable(rel_id, &oid_del_info);
    }
    RemoveImcstore(rel, oid_del_info, RELATION_IS_PARTITIONED(rel));
    heap_close(rel, NoLock);

    pq_putemptymessage('O'); /* PlanIdComplete */
    pq_flush();
    finish_xact_command();
}

void RemovePartitionedImcstoreOnStandby(Oid rel_id, StringInfo input_message)
{
    errno_t rc = EOK;

    Oid part_id = InvalidOid;
    rc = memcpy_s(&part_id, sizeof(Oid), pq_getmsgbytes(input_message, sizeof(Oid)), sizeof(Oid));
    securec_check(rc, "\0", "\0");

    Oid subpart_id = InvalidOid;
    rc = memcpy_s(&subpart_id, sizeof(Oid), pq_getmsgbytes(input_message, sizeof(Oid)), sizeof(Oid));
    securec_check(rc, "\0", "\0");

    std::unordered_map<Oid, RelFileNode> *oid_del_info = nullptr;

    start_xact_command();
    Relation rel = heap_open(rel_id, NoLock);
    if (RelationIsSubPartitioned(rel)) {
        ListCell *cell = NULL;
        List *subpart_oid_list = PartOidGetSubPartitionOidList(rel, part_id, NoLock); // subpartition oid list
        if (subpart_id != InvalidOid) {
            Partition subpart = partitionOpen(rel, subpart_id, NoLock);
            Relation part_rel = partitionGetRelation(rel, subpart);
            /* drop delta table */
            DropImcDeltaTable(subpart_id, &oid_del_info);
            RemoveImcstore(part_rel, oid_del_info, false);
            releaseDummyRelation(&part_rel);
            partitionClose(rel, subpart, NoLock);
        } else {
            foreach (cell, subpart_oid_list) {
                Oid subpart_oid = lfirst_oid(cell);
                Oid imcs_oid = search_table_from_gs_imcs(subpart_oid);
                if (OidIsValid(imcs_oid)) {
                    Partition subpart = partitionOpen(rel, subpart_oid, NoLock);
                    Relation part_rel = partitionGetRelation(rel, subpart);
                    /* drop delta table */
                    DropImcDeltaTable(subpart_oid, &oid_del_info);
                    RemoveImcstore(part_rel, oid_del_info, false);
                    releaseDummyRelation(&part_rel);
                    partitionClose(rel, subpart, NoLock);
                }
            }
        }
    } else {
        Partition part = partitionOpen(rel, part_id, NoLock);
        Relation part_rel = partitionGetRelation(rel, part);
        /* drop delta table */
        DropImcDeltaTable(part_id, &oid_del_info);
        RemoveImcstore(part_rel, oid_del_info, false);
        releaseDummyRelation(&part_rel);
        partitionClose(rel, part, NoLock);
    }

    heap_close(rel, NoLock);
    finish_xact_command();
    pq_putemptymessage('O'); /* PlanIdComplete */
    pq_flush();
}

static void HandlePgxcReceive(int conn_count, PGXCNodeHandle **temp_connections)
{
    RemoteQueryState *combiner = NULL;
    combiner = CreateResponseCombiner(conn_count, COMBINE_TYPE_NONE);
    while (conn_count > 0) {
        if (pgxc_node_receive(conn_count, temp_connections, NULL)) {
            int error_code;
            char *error_msg = getSocketError(&error_code);

            ereport(ERROR, (errcode(error_code),
                errmsg("Failed to read response from Datanodes while sending rel_id for HTAP populate. Detail: %s\n",
                error_msg)));
        }
        int i = 0;
        while (i < conn_count) {
            int res = HandleImcsResponse(temp_connections[i], combiner);
            if (res == RESPONSE_EOF) {
                i++;
            } else if (res == RESPONSE_PLANID_OK) {
                if (--conn_count > i)
                    temp_connections[i] = temp_connections[conn_count];
            } else {
                temp_connections[i]->state = DN_CONNECTION_STATE_ERROR_FATAL;
                ereport(ERROR, (errcode(ERRCODE_CONNECTION_EXCEPTION),
                    errmsg("Unexpected response from %s while sending rel_id for HTAP populate",
                    temp_connections[i]->remoteNodeName),
                    errdetail("%s", (combiner->errorMessage == NULL) ? "none" : combiner->errorMessage)));
            }
        }
        /* report error if any */
        pgxc_node_report_error(combiner);
    }
    ValidateAndCloseCombiner(combiner);
    pfree_ext(temp_connections);
}

static void PackBasicImcstoredRequest(PGXCNodeHandle *temp_connection, int msglen, Oid rel_id, List *columnlist,
    int type)
{
    errno_t ss_rc = EOK;
    /* msgType + msgLen */
    ensure_out_buffer_capacity(1 + msglen, temp_connection);
    Assert(temp_connection->outBuffer != NULL);
    temp_connection->outBuffer[temp_connection->outEnd++] = 'x';
    msglen = htonl(msglen);
    ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
        temp_connection->outSize - temp_connection->outEnd - 1, &msglen, sizeof(int));
    securec_check(ss_rc, "\0", "\0");
    temp_connection->outEnd += 4;

    ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
        temp_connection->outSize - temp_connection->outEnd - 1, &type, sizeof(int));
    securec_check(ss_rc, "\0", "\0");
    temp_connection->outEnd += 4;

    ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
        temp_connection->outSize - temp_connection->outEnd, &rel_id, sizeof(Oid));
    securec_check(ss_rc, "\0", "\0");
    temp_connection->outEnd += sizeof(Oid);

    int col_count = list_length(columnlist);
    if (col_count > 0) {
        ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
            temp_connection->outSize - temp_connection->outEnd, &col_count, sizeof(int));
        securec_check(ss_rc, "\0", "\0");
        temp_connection->outEnd += sizeof(int);
        ListCell *cell;
        foreach (cell, columnlist) {
            char *col_name = strVal(lfirst(cell));
            int name_len = strlen(col_name) + 1;
            ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
                temp_connection->outSize - temp_connection->outEnd, col_name, name_len);
            securec_check(ss_rc, "\0", "\0");
            temp_connection->outEnd += name_len;
        }
        XLogRecPtr current_lsn = t_thrd.shemem_ptr_cxt.XLogCtl->LogwrtRqst.Write;
        ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
            temp_connection->outSize - temp_connection->outEnd, &current_lsn, sizeof(XLogRecPtr));
        securec_check(ss_rc, "\0", "\0");
        temp_connection->outEnd += sizeof(XLogRecPtr);
    }
}

void SendImcstoredRequest(PGXCNodeHandle **connections, int conn_count, Oid rel_id, List *columnlist, int length)
{
    PGXCNodeHandle **temp_connections = NULL;
    /* use temp connections instead */
    int i = 0;
    temp_connections = (PGXCNodeHandle **)palloc(conn_count * sizeof(PGXCNodeHandle *));
    for (i = 0; i < conn_count; i++)
        temp_connections[i] = connections[i];

    for (i = 0; i < conn_count; i++) {
        int msglen = sizeof(int) + sizeof(int) + sizeof(Oid) + sizeof(int) + length + sizeof(XLogRecPtr);

        if (temp_connections[i]->state == DN_CONNECTION_STATE_QUERY)
            BufferConnection(temp_connections[i]);

        if (connections[i]->state != DN_CONNECTION_STATE_IDLE)
            LIBCOMM_DEBUG_LOG("Populate failed, send rel_id to node:%s[nid:%hu,sid:%hu] with abnormal state:%d",
                temp_connections[i]->remoteNodeName, temp_connections[i]->gsock.idx, temp_connections[i]->gsock.sid,
                temp_connections[i]->state);

        PackBasicImcstoredRequest(temp_connections[i], msglen, rel_id, columnlist, TYPE_IMCSTORED);

        if (pgxc_node_flush(temp_connections[i]) != 0) {
            temp_connections[i]->state = DN_CONNECTION_STATE_ERROR_FATAL;
            ereport(ERROR, (errcode(ERRCODE_CONNECTION_EXCEPTION),
                errmsg("Failed to send populate request to %s", temp_connections[i]->remoteNodeName)));
        }

        temp_connections[i]->state = DN_CONNECTION_STATE_QUERY;
    }

    HandlePgxcReceive(conn_count, temp_connections);
}

void SendPartitionedImcstoredRequest(PGXCNodeHandle **connections, int conn_count, Oid rel_id, Oid part_id,
    Oid subpart_id, List *columnlist, int length)
{
    PGXCNodeHandle **temp_connections = NULL;
    /* use temp connections instead */
    int i = 0;
    errno_t ss_rc = 0;
    temp_connections = (PGXCNodeHandle **)palloc(conn_count * sizeof(PGXCNodeHandle *));
    for (i = 0; i < conn_count; i++)
        temp_connections[i] = connections[i];

    for (i = 0; i < conn_count; i++) {
        int msglen = sizeof(int) + sizeof(int) + sizeof(Oid) + sizeof(Oid) + sizeof(Oid) + sizeof(int) + length +
            sizeof(XLogRecPtr);

        if (temp_connections[i]->state == DN_CONNECTION_STATE_QUERY)
            BufferConnection(temp_connections[i]);

        if (connections[i]->state != DN_CONNECTION_STATE_IDLE)
            LIBCOMM_DEBUG_LOG("Populate failed, send rel_id to node:%s[nid:%hu,sid:%hu] with abnormal state:%d",
                temp_connections[i]->remoteNodeName, temp_connections[i]->gsock.idx, temp_connections[i]->gsock.sid,
                temp_connections[i]->state);

        PackBasicImcstoredRequest(temp_connections[i], msglen, rel_id, columnlist, TYPE_PARTITION_IMCSTORED);

        ss_rc = memcpy_s(temp_connections[i]->outBuffer + temp_connections[i]->outEnd,
            temp_connections[i]->outSize - temp_connections[i]->outEnd, &part_id, sizeof(Oid));
        securec_check(ss_rc, "\0", "\0");
        temp_connections[i]->outEnd += sizeof(Oid);

        ss_rc = memcpy_s(temp_connections[i]->outBuffer + temp_connections[i]->outEnd,
            temp_connections[i]->outSize - temp_connections[i]->outEnd, &subpart_id, sizeof(Oid));
        securec_check(ss_rc, "\0", "\0");
        temp_connections[i]->outEnd += sizeof(Oid);

        if (pgxc_node_flush(temp_connections[i]) != 0) {
            temp_connections[i]->state = DN_CONNECTION_STATE_ERROR_FATAL;
            ereport(ERROR, (errcode(ERRCODE_CONNECTION_EXCEPTION),
                errmsg("Failed to send populate request to %s", temp_connections[i]->remoteNodeName)));
        }

        temp_connections[i]->state = DN_CONNECTION_STATE_QUERY;
    }

    HandlePgxcReceive(conn_count, temp_connections);
}

void SendUnImcstoredRequest(PGXCNodeHandle **connections, int conn_count, Oid rel_id)
{
    PGXCNodeHandle **temp_connections = NULL;
    /* use temp connections instead */
    int i = 0;

    temp_connections = (PGXCNodeHandle **)palloc(conn_count * sizeof(PGXCNodeHandle *));
    for (i = 0; i < conn_count; i++)
        temp_connections[i] = connections[i];

    for (i = 0; i < conn_count; i++) {
        int msglen = sizeof(int) + sizeof(int) + sizeof(Oid);

        if (temp_connections[i]->state == DN_CONNECTION_STATE_QUERY)
            BufferConnection(temp_connections[i]);

        if (connections[i]->state != DN_CONNECTION_STATE_IDLE)
            LIBCOMM_DEBUG_LOG("Unpopulate failed, send rel_id to node:%s[nid:%hu,sid:%hu] with abnormal state:%d",
                temp_connections[i]->remoteNodeName, temp_connections[i]->gsock.idx, temp_connections[i]->gsock.sid,
                temp_connections[i]->state);

        PackBasicImcstoredRequest(temp_connections[i], msglen, rel_id, NULL, TYPE_UNIMCSTORED);

        if (pgxc_node_flush(temp_connections[i]) != 0) {
            temp_connections[i]->state = DN_CONNECTION_STATE_ERROR_FATAL;
            ereport(ERROR, (errcode(ERRCODE_CONNECTION_EXCEPTION),
                errmsg("Failed to send unpopulate request to %s", temp_connections[i]->remoteNodeName)));
        }

        temp_connections[i]->state = DN_CONNECTION_STATE_QUERY;
    }

    HandlePgxcReceive(conn_count, temp_connections);
}

void SendPartitionedUnImcstoredRequest(PGXCNodeHandle **connections, int conn_count, Oid rel_id, Oid part_id,
    Oid subpart_id)
{
    PGXCNodeHandle **temp_connections = NULL;
    /* use temp connections instead */
    int i = 0;
    errno_t ss_rc = 0;

    temp_connections = (PGXCNodeHandle **)palloc(conn_count * sizeof(PGXCNodeHandle *));
    for (i = 0; i < conn_count; i++)
        temp_connections[i] = connections[i];

    for (i = 0; i < conn_count; i++) {
        int msglen = sizeof(int) + sizeof(int) + sizeof(Oid) + sizeof(Oid) + sizeof(Oid);

        if (temp_connections[i]->state == DN_CONNECTION_STATE_QUERY)
            BufferConnection(temp_connections[i]);

        if (connections[i]->state != DN_CONNECTION_STATE_IDLE)
            LIBCOMM_DEBUG_LOG("Unpopulate failed, send rel_id to node:%s[nid:%hu,sid:%hu] with abnormal state:%d",
                temp_connections[i]->remoteNodeName, temp_connections[i]->gsock.idx, temp_connections[i]->gsock.sid,
                temp_connections[i]->state);

        PackBasicImcstoredRequest(temp_connections[i], msglen, rel_id, NULL, TYPE_PARTITION_UNIMCSTORED);

        ss_rc = memcpy_s(temp_connections[i]->outBuffer + temp_connections[i]->outEnd,
            temp_connections[i]->outSize - temp_connections[i]->outEnd, &part_id, sizeof(Oid));
        securec_check(ss_rc, "\0", "\0");
        temp_connections[i]->outEnd += sizeof(Oid);

        ss_rc = memcpy_s(temp_connections[i]->outBuffer + temp_connections[i]->outEnd,
            temp_connections[i]->outSize - temp_connections[i]->outEnd, &subpart_id, sizeof(Oid));
        securec_check(ss_rc, "\0", "\0");
        temp_connections[i]->outEnd += sizeof(Oid);

        if (pgxc_node_flush(temp_connections[i]) != 0) {
            temp_connections[i]->state = DN_CONNECTION_STATE_ERROR_FATAL;
            ereport(ERROR, (errcode(ERRCODE_CONNECTION_EXCEPTION),
                errmsg("Failed to send unpopulate request to %s", temp_connections[i]->remoteNodeName)));
        }

        temp_connections[i]->state = DN_CONNECTION_STATE_QUERY;
    }

    HandlePgxcReceive(conn_count, temp_connections);
}

// copy void InitMultinodeExecutor(bool is_force)
PGXCNodeHandle *InitMultiNodeExecutor(Oid nodeoid)
{
    PGXCNodeHandle *result = (PGXCNodeHandle *)palloc0(sizeof(PGXCNodeHandle));
    result->sock = NO_SOCKET;
    init_pgxc_handle(result);
    result->nodeoid = nodeoid;
    result->remote_node_type = VDATANODE;
    return result;
}

PGXCNodeHandle **GetStandbyConnections(int *conn_count)
{
    int dn_conn_count = MAX_REPLNODE_NUM;
    PGXCNodeHandle **connections = (PGXCNodeHandle **)palloc(dn_conn_count * sizeof(PGXCNodeHandle *));
    Oid *dnNode = (Oid *)palloc0(sizeof(Oid) * dn_conn_count);
    char **connectionStrs = (char **)palloc0(sizeof(char *) * dn_conn_count);
    PGconn **nodeCons = (PGconn **)palloc0(sizeof(PGconn *) * dn_conn_count);
    errno_t rc;
    int repl_arr_length;
    auto release_connect = [&](char *err_msg) {
        for (int i = 0; i < dn_conn_count; i++) {
            pfree_ext(connectionStrs[i]);
        }
        pfree_ext(dnNode);
        pfree_ext(connectionStrs);
        if (err_msg != NULL) {
            pfree_ext(connections);
            connections = NULL;
            ereport(ERROR, (errmsg("PQconnectdbParallel error: %s", err_msg)));
        }
        return;
    };

    for (int i = 1; i < MAX_REPLNODE_NUM; i++) {
        char *replconninfo = NULL;
        replconninfo = u_sess->attr.attr_storage.ReplConnInfoArr[i];
        ReplConnInfo *repl_conn_info = GetReplConnInfo(replconninfo, &repl_arr_length);

        if (repl_conn_info == NULL)
            continue;

        connectionStrs[i - 1] = (char *)palloc0(INITIAL_EXPBUFFER_SIZE * 4);
        NodeDefinition *node = (NodeDefinition *)palloc0(sizeof(NodeDefinition));
        rc = strncpy_s(node->nodehost.data, NAMEDATALEN, repl_conn_info->remotehost, NAMEDATALEN);
        securec_check_c(rc, "\0", "\0");
        rc = strncpy_s(node->nodehost1.data, NAMEDATALEN, repl_conn_info->remotehost, NAMEDATALEN);
        securec_check_c(rc, "\0", "\0");
        node->nodeport = repl_conn_info->remoteport;
        sprintf_s(connectionStrs[i - 1], INITIAL_EXPBUFFER_SIZE * 4,
            "host=%s port=%d dbname=%s user=%s application_name=coordinator1 connect_timeout=600 rw_timeout=600 \
        options='-c remotetype=coordinator  -c DateStyle=iso,mdy -c timezone=prc -c geqo=on -c intervalstyle=postgres \
        -c lc_monetary=en_US.UTF-8 -c lc_numeric=en_US.UTF-8	-c lc_time=en_US.UTF-8 -c omit_encoding_error=off' \
        prototype=1 keepalives_idle=600 keepalives_interval=30 keepalives_count=20 \
        backend_version=%u enable_ce=1",
            node->nodehost.data, node->nodeport, u_sess->proc_cxt.MyProcPort->database_name,
            u_sess->proc_cxt.MyProcPort->user_name, GRAND_VERSION_NUM);
        dnNode[i - 1] = node->nodeoid;
        connections[i - 1] = InitMultiNodeExecutor(node->nodeoid);
        connections[i - 1]->nodeIdx = i;
        *conn_count = *conn_count + 1;
    }

    PQconnectdbParallel(connectionStrs, dn_conn_count, nodeCons, dnNode);

    // ListCell *node_list_item = NULL;
    for (int i = 0; i < *conn_count; i++) {
        if (nodeCons[i] && (CONNECTION_OK == nodeCons[i]->status)) {
            pgxc_node_init(connections[i], nodeCons[i]->sock);
        } else {
            char firstError[INITIAL_EXPBUFFER_SIZE] = {0};
            errno_t ss_rc = EOK;
            if (nodeCons[i] == NULL) {
                ss_rc = strcpy_s(firstError, INITIAL_EXPBUFFER_SIZE, "out of memory");
            } else if (nodeCons[i]->errorMessage.data != NULL) {
                if (strlen(nodeCons[i]->errorMessage.data) >= INITIAL_EXPBUFFER_SIZE) {
                    nodeCons[i]->errorMessage.data[INITIAL_EXPBUFFER_SIZE - 1] = '\0';
                }
                ss_rc = strcpy_s(firstError, INITIAL_EXPBUFFER_SIZE, nodeCons[i]->errorMessage.data);
            } else {
                ss_rc = strcpy_s(firstError, INITIAL_EXPBUFFER_SIZE, "unknown error");
            }
            release_connect(firstError);
        }
    }
    release_connect(NULL);
    return connections;
}

/* drop delta table */
void DropImcDeltaTable(Oid rel_id, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
#ifdef ENABLE_HTAP
    HTAPDropTable(rel_id, out_oid_del_info);
#endif
}

void DropImcDeltaTableForPartitions(Oid rel_id, Relation rel, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
    Oid imcs_oid = InvalidOid;

    LOCKMODE lock = t_thrd.postmaster_cxt.HaShmData->current_mode == STANDBY_MODE ? NoLock : ShareLock;
    if (RelationIsSubPartitioned(rel)) {
        Oid part_oid;
        ListCell *cell = NULL;
        List *part_oid_list = RelationGetSubPartitionOidList(rel, lock);
        foreach (cell, part_oid_list) {
            part_oid = lfirst_oid(cell);
            imcs_oid = search_table_from_gs_imcs(part_oid);
            if (OidIsValid(imcs_oid)) {
                Partition part = partitionOpen(rel, part_oid, lock);
                Relation part_rel = partitionGetRelation(rel, part);
                DropImcDeltaTable(part_oid, out_oid_del_info);
                releaseDummyRelation(&part_rel);
                partitionClose(rel, part, NoLock);
            }
        }
    } else {
        Oid part_oid;
        ListCell *cell = NULL;
        List *part_oid_list = relationGetPartitionOidList(rel);
        foreach (cell, part_oid_list) {
            part_oid = lfirst_oid(cell);
            imcs_oid = search_table_from_gs_imcs(part_oid);
            if (OidIsValid(imcs_oid)) {
                Partition part = partitionOpen(rel, part_oid, lock);
                Relation part_rel = partitionGetRelation(rel, part);
                DropImcDeltaTable(part_oid, out_oid_del_info);
                releaseDummyRelation(&part_rel);
                partitionClose(rel, part, NoLock);
            }
        }
    }
}

static int16 *GetImcskeyFromGsImcs(HeapTuple tuple, TupleDesc imcstup_desc, int *n_column)
{
    Datum imcskey_raw;
    bool isnull = false;
    ArrayType *imcskey_columns = NULL;

    Assert(PointerIsValid(n_column));
    imcskey_raw = heap_getattr(tuple, Anum_gs_imcs_imcskey, imcstup_desc, &isnull);

    Assert(!isnull);
    imcskey_columns = DatumGetArrayTypeP(imcskey_raw);
    *n_column = ARR_DIMS(imcskey_columns)[0];

    if (ARR_NDIM(imcskey_columns) != 1 || *n_column < 0 || ARR_HASNULL(imcskey_columns) ||
        ARR_ELEMTYPE(imcskey_columns) != INT2OID) {
        ereport(ERROR,
            (errcode(ERRCODE_ARRAY_ELEMENT_ERROR), errmsg("imcskey column's number is not a 1-D smallint array")));
    }

    /* Get int2 array of index key column numbers */
    return (int16 *)ARR_DATA_PTR(imcskey_columns);
}

int2vector *FormImcsKeysFromGsImcs(HeapTuple tup, TupleDesc imcstup_desc)
{
    int16 *attnum = NULL;
    int2vector *imcskey = NULL;
    int n_column = 0;
    attnum = GetImcskeyFromGsImcs(tup, imcstup_desc, &n_column);

    imcskey = buildint2vector(NULL, n_column);
    for (int i = 0; i < n_column; i++) {
        imcskey->values[i] = attnum[i];
    }
    return imcskey;
}

int2vector *GetImcsKeys(Oid rel_oid)
{
    ServerMode current_mode = t_thrd.postmaster_cxt.HaShmData->current_mode;
    if (current_mode == STANDBY_MODE) {
        bool found = false;
        IMCSHashEntry *result_found = NULL;
        result_found =
            (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &rel_oid, HASH_FIND, &found);
        if (!found) {
            return nullptr;
        }
        int imcs_natts = result_found->imcsnattr;
        int2vector *imcskey = (int2vector *)palloc0(Int2VectorSize(imcs_natts));
        errno_t rc = memcpy_s(imcskey, Int2VectorSize(imcs_natts), result_found->imcskey, Int2VectorSize(imcs_natts));
        securec_check(rc, "\0", "\0");
        return imcskey;
    } else {
        Oid imcs_oid = search_table_from_gs_imcs(rel_oid);
        if (imcs_oid == InvalidOid) {
            return nullptr;
        }

        Relation relation = NULL;
        TableScanDesc scan;
        Form_gs_imcs gsimcs_form;
        HeapTuple heap_tup = NULL;
        TupleDesc imcstup_desc;

        /* search imcs_oid in gs_imcs catalog */
        relation = heap_open(ImcsRelationId, RowExclusiveLock);
        imcstup_desc = RelationGetDescr(relation);
        scan = tableam_scan_begin(relation, SnapshotNow, 0, NULL);
        heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
        while (heap_tup != NULL) {
            gsimcs_form = (Form_gs_imcs)GETSTRUCT(heap_tup);
            if (gsimcs_form->imcsoid == imcs_oid) {
                /* found tuple */
                break;
            }
            heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
        }

        if (!HeapTupleIsValid(heap_tup)) {
            heap_endscan(scan);
            heap_close(relation, NoLock);
            return nullptr;
        }

        int2vector *imcskey = FormImcsKeysFromGsImcs(heap_tup, imcstup_desc);

        heap_endscan(scan);
        heap_close(relation, NoLock);
        return imcskey;
    }
}

void ImcsNameFormRel(Oid rel_oid, char *imcs_meta_name, char *imcs_meta_index_name)
{
    errno_t rc = EOK;
    rc = snprintf_s(imcs_meta_name, NAMEDATALEN, NAMEDATALEN - 1, "gs_imcudesc_%u", rel_oid);
    securec_check_ss(rc, "", "");
    rc = snprintf_s(imcs_meta_index_name, NAMEDATALEN, NAMEDATALEN - 1, "gs_imcudesc_%u_index", rel_oid);
    securec_check_ss(rc, "", "");
}

static inline bool CheckUnImcstoredCol(CreateImcsStmt *imcs_stmt, const char *attname)
{
    ListCell *colcell = NULL;
    ImcsElem *elem = NULL;
    foreach (colcell, imcs_stmt->unimcstored_columns) {
        elem = (ImcsElem *)lfirst(colcell);
        if ((pg_strcasecmp(elem->colname, attname) == 0) && elem->is_imcstored == false) {
            return true;
        }
    }
    return false;
}

static inline bool CheckTableEnableImcs(CreateImcsStmt *imcs_stmt)
{
    if (imcs_stmt == NULL || imcs_stmt->relkind != RELKIND_IMCS) {
        return false;
    }

    if (!imcs_stmt->table_has_imcs && !imcs_stmt->imcstored_columns && !imcs_stmt->unimcstored_columns) {
        return false;
    }

    return true;
}


Oid CreateImcsTableOnRel(CreateImcsStmt *imcs_stmt, bool is_partition, Oid parent_oid, Oid rel_oid, Relation rel)
{
    Oid imcsmeta_oid = InvalidOid;
    char *imcs_meta_name = NULL;
    char *imcs_meta_index_name = NULL;
    Datum reloptions = (Datum)0;
    Oid collation_object_id[2];
    Oid class_object_id[2];
    int16 coloptions[2];
    Oid index_oid = InvalidOid;
    char *rel_name = NULL;

    if (!CheckTableEnableImcs(imcs_stmt) && !is_partition) {
        return InvalidOid;
    }
    /*
    is_partition: rel is a partition table or not
    has_partition：it is a part or subpart
    */
    if (is_partition) {
        rel_name = getPartitionName(rel_oid, true);
    } else {
        rel_name = get_rel_name(rel_oid);
    }
    bool has_partition = RELATION_IS_PARTITIONED(rel) || is_partition;

    imcs_meta_name = (char *)palloc0(sizeof(char) * NAMEDATALEN);
    imcs_meta_index_name = (char *)palloc0(sizeof(char) * NAMEDATALEN);

    ImcsNameFormRel(rel_oid, imcs_meta_name, imcs_meta_index_name);

    /*
     * If we add one column, we should change const variable
     * Natts_gs_imcu_descs in gs_imcs.h
     */
    TupleDesc tupdesc = CreateTemplateTupleDesc(Natts_gs_imcu_descs, false);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_reloid, "imcsoid", OIDOID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_xmin, "imu_xmin", XIDOID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_id, "imcu_id", INT4OID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_min, "imcu_min", NAMEOID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_max, "imcu_max", NAMEOID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_row_count, "row_count", INT4OID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_size, "imcu_size", INT4OID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_mode, "imcu_mode", INT4OID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_pointer, "imcu_pointer", INT8OID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_magic, "magic", INT4OID, -1, 0);

    /* add internal_mask_enable */
    reloptions = AddInternalOption(reloptions, INTERNAL_MASK_DALTER | INTERNAL_MASK_DDELETE | INTERNAL_MASK_DINSERT |
        INTERNAL_MASK_DUPDATE | INTERNAL_MASK_ENABLE);

    char relkind = RELKIND_IMCS;

    imcsmeta_oid = heap_create_with_catalog(imcs_meta_name, rel->rd_rel->relnamespace, rel->rd_rel->reltablespace,
        InvalidOid, InvalidOid, InvalidOid, rel->rd_rel->relowner, tupdesc, NIL, relkind, rel->rd_rel->relpersistence,
        rel->rd_rel->relisshared, false, true, 0, ONCOMMIT_NOOP, reloptions, false, true, NULL, REL_CMPRS_NOT_SUPPORT,
        NULL);

    Assert(imcsmeta_oid != InvalidOid);
    /* make the imcs relation visible, else heap_open will fail */
    CommandCounterIncrement();

    /* SHARE_LOCK is not really needed here, but take it anyway */
    Relation imcudesc_rel = heap_open(imcsmeta_oid, ShareLock);

    IndexInfo *index_info = makeNode(IndexInfo);
    index_info->ii_NumIndexAttrs = 2;
    index_info->ii_NumIndexKeyAttrs = 2;
    index_info->ii_KeyAttrNumbers[0] = 1;
    index_info->ii_KeyAttrNumbers[1] = 2;
    index_info->ii_Expressions = NIL;
    index_info->ii_ExpressionsState = NIL;
    index_info->ii_Predicate = NIL;
    index_info->ii_PredicateState = NIL;
    index_info->ii_ExclusionOps = NULL;
    index_info->ii_ExclusionProcs = NULL;
    index_info->ii_ExclusionStrats = NULL;
    index_info->ii_Unique = true;
    index_info->ii_ReadyForInserts = true;
    index_info->ii_Concurrent = false;
    index_info->ii_BrokenHotChain = false;
    index_info->ii_ParallelWorkers = 0;
    index_info->ii_PgClassAttrId = Anum_pg_class_relcudescidx;
    collation_object_id[0] = InvalidOid;
    collation_object_id[1] = InvalidOid;
    class_object_id[0] = INT4_BTREE_OPS_OID;
    class_object_id[1] = INT4_BTREE_OPS_OID;
    coloptions[0] = 0;
    coloptions[1] = 0;

    IndexCreateExtraArgs extra;
    SetIndexCreateExtraArgs(&extra, InvalidOid, false, false);

    index_oid = index_create(imcudesc_rel, imcs_meta_index_name, InvalidOid, InvalidOid, index_info,
        list_make2((void *)"col_id", (void *)"cu_id"), BTREE_AM_OID, rel->rd_rel->reltablespace, collation_object_id,
        class_object_id, coloptions, (Datum)0, true, false, false, false, true, false, false, &extra, false);

    Assert(OidIsValid(index_oid));

    heap_close(imcudesc_rel, NoLock);

    /*
     * Make changes visible
     */
    CommandCounterIncrement();

    AlterTableCreateToastTable(imcsmeta_oid, reloptions);


    int i;
    int j = 1;
    TupleDesc rel_desc = rel->rd_att;
    FormData_pg_attribute *rel_atts = rel_desc->attrs;
    int rel_attnum_all = rel_desc->natts;
    int rel_attnum = rel_attnum_all;
    for (i = 1; i <= rel_attnum_all; i++) {
        if (rel_atts[i - 1].attisdropped) {
            rel_attnum--;
        }
    }

    int imcstored_col_nums = 0;
    int2vector *imcskey = NULL;
    ListCell *colcell = NULL;
    ImcsElem *elem = NULL;
    if (imcs_stmt->table_has_imcs) {
        imcstored_col_nums = rel_attnum - imcs_stmt->unimcstored_col_nums;
        imcskey = buildint2vector(NULL, imcstored_col_nums);
        for (i = 1, j = 1; i <= rel_attnum_all; i++) {
            /* add IMCSTORED columns only */
            if (!(rel_atts[i - 1].attisdropped) && !CheckUnImcstoredCol(imcs_stmt, rel_atts[i - 1].attname.data)) {
                imcskey->values[j - 1] = rel_atts[i - 1].attnum;
                j++;
            }
        }
    } else {
        imcstored_col_nums = imcs_stmt->imcstored_col_nums;
        imcskey = buildint2vector(NULL, imcstored_col_nums);
        foreach (colcell, imcs_stmt->imcstored_columns) {
            elem = (ImcsElem *)lfirst(colcell);
            imcskey->values[j - 1] = attnameAttNum(rel, elem->colname, false);
            j++;
        }
    }

    gs_imcs_insert_tuple(rel_oid, rel_name, imcsmeta_oid, imcs_meta_name, has_partition, parent_oid, imcstored_col_nums,
        imcskey);

    if (imcs_stmt->baseTableNotcreated == true) {
        UpdateImcHashTbl(rel_oid, rel->rd_att->natts, imcskey->values, imcstored_col_nums);
        if (has_partition == true) {
            if (rel_oid == parent_oid) {
                UpdateImcHashTbl(rel_oid, rel->rd_att->natts, imcskey->values, imcstored_col_nums, false);
            } else {
                Oid base_rel_id = rel->grandparentId == InvalidOid ? parent_oid : rel->grandparentId;
                Relation baserel = heap_open(base_rel_id, NoLock);
                if (RelationIsSubPartitioned(baserel) &&
                    (rel->parentId != InvalidOid && rel->grandparentId == InvalidOid)) {
                    UpdateImcHashTbl(rel_oid, rel->rd_att->natts, imcskey->values, imcstored_col_nums, false);
                }
                heap_close(baserel, NoLock);
            }
        }
    }

    if (is_partition) {
        CacheInvalidatePartcacheByPartid(rel_oid); // partition oid
    } else {
        CacheInvalidateRelcacheByRelid(rel_oid);
    }
    CacheInvalidateRelcacheByRelid(imcsmeta_oid);
    /*
     * Make changes visible
     */
    CommandCounterIncrement();

    pfree_ext(imcskey);
    pfree_ext(rel_name);

    return imcsmeta_oid;
}

/* delete heap tuple from pg_class, pg_attribute and gs_imcs for imcs */
void DeleteImcsTableOnRel(Oid rel_oid)
{
    Relation rel = InvalidOid;
    std::unordered_map<Oid, RelFileNode> *oid_del_info = nullptr;

    if (!OidIsValid(rel_oid)) {
        return;
    }

    rel = heap_open(rel_oid, AccessExclusiveLock);

    /* drop delta table */
    if (RELATION_IS_PARTITIONED(rel)) {
        DropImcDeltaTableForPartitions(rel_oid, rel, &oid_del_info);
    } else {
        DropImcDeltaTable(rel_oid, &oid_del_info);
    }

    RemoveImcstore(rel, oid_del_info, RELATION_IS_PARTITIONED(rel));
    delete_gs_imcs_table_on_rel(rel, oid_del_info);
    heap_close(rel, AccessExclusiveLock);
    return;
}

/**
 * check whether relation support imcstore
 * @param relation relation
 * @return
 */
bool CheckImcsSupportForRelType(Relation relation)
{
    RelFileCompressOption current;
    TransCompressOptions(relation->rd_node,&current);
    if (IsSystemRelation(relation) || IsCatalogRelation(relation) || IsToastRelation(relation) ||
        RelationIsToast(relation) || isAnyTempNamespace(RelationGetNamespace(relation)) || RELATION_IS_TEMP(relation) ||
        RelationGetRelPersistence(relation) == RELPERSISTENCE_UNLOGGED || RowRelationIsCompressed(relation) ||
        relation->is_compressed || relation->rd_isblockchain ||
        relation->storage_type != HEAP_DISK || current.compressLevelSymbol != 0) {
        ereport(ERROR, (errmsg("Table type not support for Imcstore.")));
        return false;
    }
    return true;
}

/* IMCS */
